package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseAppV2;
import com.atguigu.gmall.realtime.bean.OrderDetail;
import com.atguigu.gmall.realtime.bean.OrderInfo;
import com.atguigu.gmall.realtime.bean.OrderWide;
import com.atguigu.gmall.realtime.util.MyJDBCUtil;
import com.atguigu.gmall.realtime.util.MyRedisUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.time.Duration;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/3/17 15:16
 */
public class DWMOrderWideApp_Cache extends BaseAppV2 {
    
    public static void main(String[] args) {
        new DWMOrderWideApp_Cache().init(20003, 2, "DWMOrderWideApp", "DWMOrderWideApp", "dwd_order_info", "dwd_order_detail");
    }
    
    @Override
    public void run(StreamExecutionEnvironment env,
                    Map<String, DataStreamSource<String>> sourceStreams) {
        // 1. 订单表和订单事实表进行join
        final SingleOutputStreamOperator<OrderWide> orderWideWithoutDimStream = factTableJoin(sourceStreams);
        // 2. join维度
        joinDim(orderWideWithoutDimStream);
    }
    
    private void joinDim(SingleOutputStreamOperator<OrderWide> orderWideWithoutDimStream) {
        String phoenixUrl = "jdbc:phoenix:hadoop162,hadoop163,hadoop164:2181";
        /*
            流中的数据现在是缺少维度信息, 拿到每个数据之后, 去读hbase(Phoenix), 然后不去对应的维度信息
         */
        orderWideWithoutDimStream
            .map(new RichMapFunction<OrderWide, OrderWide>() {
                private Jedis redisClient;
                private Connection conn;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    conn = MyJDBCUtil.getConnection(phoenixUrl);
                    redisClient = MyRedisUtil.getRedisClient();
                }
                
                // 读取维度
                private JSONObject readDim(String tableName, Object[] args) {
                    
                    // 先从缓存读数据, 缓存没有, 再从HBase读数据
                    String key = tableName + ":" + args[0];
                    if (redisClient.exists(key)) {
                        System.out.println(tableName + "的 id " + args[0] + " 走缓存");
                        String userInfoJson = redisClient.get(key);
                        return JSON.parseObject(userInfoJson);
                    } else {
                        System.out.println(tableName + "的 id " + args[0] + " 走数据库");
                        String sql = "select * from " + tableName + " where ID = ?";
                        JSONObject jsonObj = MyJDBCUtil
                            .queryList(conn, sql, args, JSONObject.class, false)
                            .get(0);
                        // 把数据存入缓存, 过期时间一周
                        redisClient.setex(key, 60 * 60 * 24 * 7, jsonObj.toJSONString());
                        return jsonObj;
                    }
                }
                
                @Override
                public OrderWide map(OrderWide orderWide) throws Exception {
                    // 1. join user_info 根据user_id 去查找这个use的相关信息  select * from dim_user_info where id=...
                    JSONObject userObj = readDim("DIM_USER_INFO", new Object[]{orderWide.getUser_id().toString()});
                    orderWide.calcUserAge(userObj.getString("BIRTHDAY"));
                    orderWide.setUser_gender(userObj.getString("GENDER"));  // shift+ctrl+u 大小写转换
                    
                    // 2. join 省份维度
                    JSONObject provinceObj = readDim("DIM_BASE_PROVINCE", new Object[]{orderWide.getProvince_id().toString()});
                    orderWide.setProvince_name(provinceObj.getString("NAME"));
                    orderWide.setProvince_iso_code(provinceObj.getString("ISO_CODE"));
                    orderWide.setProvince_area_code(provinceObj.getString("AREA_CODE"));
                    orderWide.setProvince_3166_2_code(provinceObj.getString("ISO_3166_2"));
                    
                    // 3. join sku维度
                    final Object[] skuArgs = {orderWide.getSku_id().toString()};
                    final JSONObject skuObj = readDim("DIM_SKU_INFO", skuArgs);
                    orderWide.setSku_name(skuObj.getString("SKU_NAME"));
                    orderWide.setSpu_id(skuObj.getLong("SPU_ID"));
                    orderWide.setCategory3_id(skuObj.getLong("CATEGORY3_ID"));
                    orderWide.setTm_id(skuObj.getLong("TM_ID"));
                    
                    // 4. join spu维度
                    final Object[] spuArgs = {orderWide.getSpu_id().toString()};
                    final JSONObject spuObj = readDim("DIM_SPU_INFO", spuArgs);
                    orderWide.setSpu_name(spuObj.getString("SPU_NAME"));
                    
                    // 5. join category3维度
                    final Object[] category3Args = {orderWide.getCategory3_id().toString()};
                    final JSONObject category3Obj = readDim("DIM_BASE_CATEGORY3", category3Args);
                    orderWide.setCategory3_name(category3Obj.getString("NAME"));
                    // 6. join 品牌维度
                    final Object[] tmArgs = {orderWide.getTm_id().toString()};
                    final JSONObject tmObj = readDim("DIM_BASE_TRADEMARK", tmArgs);
                    orderWide.setTm_name(tmObj.getString("TM_NAME"));
                    
                    // 调试
                    if (tmObj.getString("TM_NAME").contains("苹果")) {
                        System.out.println(tmObj);
                    }
                    return orderWide;
                }
                
                @Override
                public void close() throws Exception {
                    if (conn != null && !conn.isClosed()) {
                        conn.close();
                    }
                }
            })
            .print();
    }
    
    private SingleOutputStreamOperator<OrderWide> factTableJoin(Map<String, DataStreamSource<String>> sourceStreams) {
        final SingleOutputStreamOperator<OrderInfo> orderInfoStream = sourceStreams
            .get("dwd_order_info")
            .map(line -> JSON.parseObject(line, OrderInfo.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderInfo>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((orderInfo, ts) -> orderInfo.getCreate_ts())
            );
        
        final SingleOutputStreamOperator<OrderDetail> orderDetailStream = sourceStreams
            .get("dwd_order_detail")
            .map(line -> JSON.parseObject(line, OrderDetail.class))
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<OrderDetail>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                    .withTimestampAssigner((orderDetail, ts) -> orderDetail.getCreate_ts())
            );
        
        // 1. 必须使用event-time 2. 必须按照key作为join条件
        return orderInfoStream
            .keyBy(OrderInfo::getId)
            .intervalJoin(orderDetailStream.keyBy(OrderDetail::getOrder_id))
            .between(Time.seconds(-5), Time.seconds(5))
            .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                @Override
                public void processElement(OrderInfo left,
                                           OrderDetail right,
                                           Context ctx,
                                           Collector<OrderWide> out) throws Exception {
                    out.collect(new OrderWide(left, right));
                }
            });
        
    }
}
/*
缓存什么数据:
    维度表数据 (6张表数据)
缓存到位置:
    redis
redis的数据结构的选择:
    string  list set hash zset
    
    比如: 根据用户id来读取该用户的数据
    
    "dim_user_info"    "user_1"
    
    类型使用字符串:
        key                         value
        "dim_user_info:user_1"      "json格式字符串"
        
        换处: key特别多
        
    set集合: 不行
        "dim_user_info"           Set(json格式字符串, ...)
        
    
    hash                        (map)
    "dim_user_info"             "user_1":  json格式字符串
    
    
    
    



 */